package com.kunda.netty.websocket;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.CharsetUtil;

import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class WebSocketServerHandler  extends SimpleChannelInboundHandler<Object>{
	private static final Logger logger = LoggerFactory.getLogger(WebSocketServerHandler.class);
	private WebSocketServerHandshaker handsharker;
	
	@Override
	protected void messageReceived(ChannelHandlerContext ctx, Object msg)
			throws Exception {
		if(msg instanceof FullHttpRequest){
			handleHttpRequest(ctx, (FullHttpRequest)msg);
		}else if(msg instanceof WebSocketFrame){
			handleWebSocketFrame(ctx, (WebSocketFrame)msg);
		}
	}
	
	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		ctx.flush();
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
		cause.printStackTrace();
		ctx.close();
	}

	private void handleHttpRequest(ChannelHandlerContext ctx,FullHttpRequest request) throws Exception{
		if(!request.getDecoderResult().isSuccess() ||
		   (!"websocket".equals(request.headers().get("upgrade")))){
			
			FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST);
			
			sendHttpResponse(ctx, request, response);
			
			return;
			
		}
		
		String webSocketURL = "ws://localhost:8080/websocket";
		
		WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(webSocketURL, null, false);
		
		handsharker = wsFactory.newHandshaker(request);
		
		if(handsharker == null){
			WebSocketServerHandshakerFactory.sendUnsupportedWebSocketVersionResponse(ctx.channel());
		}else{
			handsharker.handshake(ctx.channel(), request);
		}
	}
	
	
	private void handleWebSocketFrame(ChannelHandlerContext ctx,WebSocketFrame frame){
		if(frame instanceof CloseWebSocketFrame){
			handsharker.close(ctx.channel(), (CloseWebSocketFrame)frame.retain());
			return ;
		}
		
		if(frame instanceof PingWebSocketFrame){
			ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
			return ;
		}
		
		if(!(frame instanceof TextWebSocketFrame)){
			throw new UnsupportedOperationException(String.format("%s frame types not support",frame.getClass().getName()));
		}
		
		String request = ((TextWebSocketFrame)frame).text();
		if(logger.isDebugEnabled()){
			logger.debug(String.format("%s received %s",ctx.channel(),request));
		}
		
		TextWebSocketFrame responseFrame = new TextWebSocketFrame(String.format("%s,welcome to Netty Websocket server,now is:%s", request,new Date()));
		ctx.channel().write(responseFrame);
	}

	
	private static void sendHttpResponse(ChannelHandlerContext ctx,FullHttpRequest request,FullHttpResponse response){
		if(response.getStatus().code()!=200){
			ByteBuf buf = Unpooled.copiedBuffer(response.getStatus().toString(),CharsetUtil.UTF_8);
			response.content().writeBytes(buf);
			buf.release();
			setContentLength(response, response.content().readableBytes());
		}
		
		ChannelFuture future = ctx.channel().writeAndFlush(response);
		if(!HttpHeaders.isKeepAlive(request) || response.getStatus().code()!=200){
			future.addListener(ChannelFutureListener.CLOSE);
		}
	}
	
	
	private static void setContentLength(HttpResponse response,long length){
		response.headers().set(HttpHeaders.Names.CONTENT_LENGTH,length);
	}
}
